package org.senRPC.client.consumer;

import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import org.senRPC.common.rpc.RegisterMeta;
import org.senRPC.common.utils.ChannelGroup;
import org.senRPC.common.utils.JUnsafe;
import org.senRPC.common.utils.UnresolvedAddress;
import org.senRPC.remoting.ConnectionUtils;
import org.senRPC.remoting.netty.NettyRemotingClient;

import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


import static java.util.concurrent.TimeUnit.MILLISECONDS;

public class SubscribeManagerImpl implements Consumer.SubscribeManager {
    private static final Logger logger = LoggerFactory.getLogger(SubscribeManagerImpl.class);
    private final ReentrantLock lock = new ReentrantLock();
    private final Condition notifyCondition = lock.newCondition();
    private final AtomicBoolean signalNeeded = new AtomicBoolean(false);
    private final String service;
    private final ConsumerManager consumerManager;
    // 持有defaultConsumer对象好办事
    private final DefaultConsumer defaultConsumer;

    public SubscribeManagerImpl(String service, ConsumerManager consumerManager, DefaultConsumer defaultConsumer) {
        this.service = service;
        this.consumerManager = consumerManager;
        this.defaultConsumer = defaultConsumer;
    }

    @Override
    public void startSubcribe() {
        if (service != null) {
            // 以下主要工作:将service与对应的Listener结合
            // 并建立与生产者的
            this.consumerManager.subcribeService(service,new NotifyListener(){
                @Override
                public void notify(RegisterMeta registerMeta, NotifyEvent event) {
                    // host
                    String remoteHost = registerMeta.getAddress().getHost();
                    // port vip服务 port端口号-2
                    int remotePort = registerMeta.isVIPService() ? (registerMeta.getAddress().getPort() - 2) : registerMeta.getAddress().getPort();
                    //因为一个服务消费者的实例和一个服务提供者之间的连接可以有多个,所以放到一个组里,又因为一个服务可能有多个实例,所以放到一个list里
                    //一个服务者的实例只有一个地址,对应着一个group.
                    final ChannelGroup group = defaultConsumer.group(new UnresolvedAddress(remoteHost, remotePort));
                    //增加信道
                    if (event == NotifyEvent.CHILD_ADDED) {
                        // 链路复用，如果此host和port对应的链接的channelGroup是已经存在的，则无需建立新的链接，只需要将此group与service建立关系即可
                        if (!group.isAvailable()) {
                            // 消费者获得了注册信息元数据,里有连接的个数
                            int connCount = registerMeta.getConnCount() < 0 ? 1 : registerMeta.getConnCount();
                            group.setWeight(registerMeta.getWeight());
                            //把所有服务提供这放到组里
                            for (int i = 0; i < connCount; i++) {
                                try {
                                    // 所有的consumer与provider之间的链接不进行短线重连操作
                                    //类名.this一般在内部类中使用
                                    NettyRemotingClient nettyRemotingClient = defaultConsumer.getProviderNettyRemotingClient();
                                    nettyRemotingClient.setreconnect(false);
                                    nettyRemotingClient.getBootstrap()
                                            //connect返回一个ChannelFuture
                                            .connect(ConnectionUtils.string2SocketAddress(remoteHost + ":" + remotePort)).addListener(new ChannelFutureListener() {
                                        // 因为在Netty中操作是异步的，ch.writeAndFlush(message)也是异步的，该方法只是把发
                                        // 送消息加入了任务队列，这时直接关闭连接会导致问题。所以我们需要在消息发送完毕后在去关闭连接。
                                        @Override
                                        //进行连接,为连接的过程,只有成功了将信道的信道加到消费者的
                                        public void operationComplete(ChannelFuture future) throws Exception {
                                            //成功之后将channel加入组中
                                            group.add(future.channel());
                                            //唤醒所有等待的线程 .多个线程只能唤醒一次,保证不重复性.即: 一个channel和注册中心的连接的channel只能唤醒一次.
                                            // getAndSet 以原子方式设置为给定值，并返回以前的值。
                                            onSucceed(signalNeeded.getAndSet(false));
                                        }
                                    });
                                } catch (Exception e) {
                                    logger.error("connection provider host [{}] and port [{}] occor exception [{}]", remoteHost, remotePort, e.getMessage());
                                }
                            }
                        }else{
                            //唤醒所有等待的线程,
                            onSucceed(signalNeeded.getAndSet(false));
                        }
                        defaultConsumer.addChannelGroup(service,group);
                    }else if(event == NotifyEvent.CHILD_REMOVED){
                        defaultConsumer.removedIfAbsent(service,group);
                    }
                }
            });
        }

    }

    @Override
    public boolean waitForAvailable(long timeoutMillis) {
        if (isServiceAvailable(service)) {
            return true;
        }
        boolean available = false;
        long start = System.nanoTime();
        final ReentrantLock _look = lock;
        _look.lock();
        try {
            while (!isServiceAvailable(service)) {
                signalNeeded.set(true);//需要被唤醒,阻塞在下面
                // 这编的不太好,如果超时了依然在阻塞中
                notifyCondition.await(timeoutMillis, MILLISECONDS);
                available = isServiceAvailable(service);
                if (available || (System.nanoTime() - start) > MILLISECONDS.toNanos(timeoutMillis)) {
                    break;
                }
            }
        } catch (InterruptedException e) {
            JUnsafe.throwException(e);
        } finally {
            _look.unlock();
        }
        return available;
    }

    private boolean isServiceAvailable(String service) {
        CopyOnWriteArrayList<ChannelGroup> list = defaultConsumer.getChannelGroupByServiceName(service);
        if(list == null){
            return false;
        }else{
            for(ChannelGroup channelGroup : list){
                //channelGroup长度大于0就可用
                if(channelGroup.isAvailable()){
                    return true;
                }
            }
        }
        return false;
    }

    private void onSucceed(boolean doSignal) {
        if (doSignal) {
            final ReentrantLock _look = lock;
            _look.lock();
            try {
                notifyCondition.signalAll();
            } finally {
                _look.unlock();
            }
        }
    }
}
